package com.zq.learn.spark.kafka;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Random;
import java.util.UUID;

/**
 * OrderProducer
 * 模拟产生订单数据,结构：
 * {"price":204,"id":"beeff9b6-1cc8-4272-96a0-7a5cd3d92358","timestamp":"20200212 17:19:58","status":0}
 */
public class OrderProducer {
  private final KafkaProducer<Integer, String> producer;
  private final String topic;
  private final Properties props = new Properties();

  public OrderProducer(String topic)
  {
    props.put("metadata.broker.list", "localhost:9092");
    props.put("bootstrap.servers", "localhost:9092");
    props.put("acks", "all");
    props.put("retries", 0);
    props.put("batch.size", 16384);
    props.put("linger.ms", 1);
    props.put("buffer.memory", 33554432);
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    producer = new KafkaProducer<Integer, String>(props);
    this.topic = topic;
  }

  public void sendMsg() {
    for(int i=0;i<100;i++){
      Map<String,Object> orderMap=new HashMap<>();
      orderMap.put("id", UUID.randomUUID().toString());
      int price = new Random().nextInt(500)+1;
      orderMap.put("price",price);
      SimpleDateFormat sdf=new SimpleDateFormat("yyyyMMdd HH:mm:ss");
      orderMap.put("timestamp",sdf.format(new Date()));
      int status = price%2;
      //status：1-成功；0-失败；
      orderMap.put("status",status);
      String orderStr = JSON.toJSONString(orderMap);
      System.out.println("Send:" + orderStr);
      producer.send(new ProducerRecord<Integer, String>(topic, orderStr));
    }
  }

}
